package com.hisense.ovcloud.mqtt.business.service;

import cn.hutool.json.JSONUtil;
import com.hisense.ovcloud.commons.configs.kafka.ReactiveKafkaSender;
import com.hisense.ovcloud.commons.dto.common.AesKeyInfoInCache;
import com.hisense.ovcloud.commons.dto.common.ProductInfoInCache;
import com.hisense.ovcloud.commons.dto.common.mqtt.MqttDeviceUploadMsg;
import com.hisense.ovcloud.commons.dto.common.mqtt.PubVerMsg;
import com.hisense.ovcloud.commons.dto.req.NotifyWifiOnlineToHitachiReq;
import com.hisense.ovcloud.commons.dto.resp.GenerateAesKeyAndFetchPublicKeyResp;
import com.hisense.ovcloud.commons.tools.AESEcbEncryptUtil;
import com.hisense.ovcloud.commons.tools.DkeyTool;
import com.hisense.ovcloud.commons.tools.LocalLFUCacheUtil;
import com.hisense.ovcloud.commons.tools.LocalTimedCacheUtil;
import com.hisense.ovcloud.mqtt.business.config.Constants;
import com.hisense.ovcloud.mqtt.business.config.MqttClientConfig;
import com.hisense.ovcloud.mqtt.business.dao.ProductDao;
import com.hisense.ovcloud.mqtt.business.entity.ProductInfo;
import com.mybatisflex.core.datasource.DataSourceKey;
import com.mybatisflex.core.row.Db;
import com.mybatisflex.core.row.RowUtil;
import io.micrometer.common.util.StringUtils;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.mqtt.MqttClient;
import io.vertx.mqtt.MqttClientOptions;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;

@Configuration
@Slf4j
@DependsOn({"sqlSessionFactory"})
public class MQTTClientConnector implements InitializingBean {

    private final MqttClientConfig mqttClientConfig;
    private final ReactiveKafkaSender reactiveKafkaSender;

    private final MqttClients mqttClients;
    private final ProductDao productDao;
    private static Scheduler kafkaUploadPool;
    private static Scheduler workerPool;

    private final WebClient webClient;

    @Value("${mqtt.client.push.qos:1}")
    private int mqttCleintPushQos;
    @Value("${mqtt.client.subscribe.qos:1}")
    private int mqttCleintSubscribeQos;

    private final Vertx vertx = Vertx.vertx();
    private final MqttClientOptions options = new MqttClientOptions()
            .setMaxMessageSize(100_000_000_0)
            .setCleanSession(true)
            .setAutoKeepAlive(true)
            .setKeepAliveInterval(2)
            .setClientId("0/0/999999999999999_1_1_" + System.currentTimeMillis() / 1000)
            .setUsername("0/0/999999999999999")
            .setPassword("0/0/999999999999999");

    private final LocalLFUCacheUtil<ProductInfoInCache> productInfoInCaches;

    private final LocalTimedCacheUtil<AesKeyInfoInCache> aesKeyInfoInCaches;
    //2. 声明一个私有的对象属性；（非线程安全的）没有volatile
    private final AtomicReference<MqttClient> client = new AtomicReference<>();

    public MQTTClientConnector(ReactiveKafkaSender reactiveKafkaSender,
                               MqttClientConfig mqttClientConfig,
                               @Qualifier("productInfoInCache") LocalLFUCacheUtil<ProductInfoInCache> productInfoInCaches,
                               ProductDao productDao, MqttClients mqttClients,
                               @Qualifier("aesKeyInfoInCache") LocalTimedCacheUtil<AesKeyInfoInCache> aesKeyInfoInCaches, @Qualifier("base") WebClient webClient) {
        this.mqttClientConfig = mqttClientConfig;
        this.reactiveKafkaSender = reactiveKafkaSender;
        this.productInfoInCaches = productInfoInCaches;
        this.productDao = productDao;
        this.mqttClients = mqttClients;
        this.aesKeyInfoInCaches = aesKeyInfoInCaches;

        options.setMaxInflightQueue(Integer.MAX_VALUE);
        options.setReconnectAttempts(Integer.MAX_VALUE);
        options.setReconnectInterval(mqttClientConfig.getReconnectInterval());
        options.setConnectTimeout(mqttClientConfig.getConnectTimeout());
        this.webClient = webClient;
    }

    public void reconnect(int checkIntervalSeconds) {
        Flux.interval(Duration.ofSeconds(checkIntervalSeconds))
                .subscribe(t -> {
                    if ((client.get() == null || !client.get().isConnected())) {
                        if(client.get() != null){
                            client.get().disconnect();
                        }
                        log.error("business is not connecting to emqx, will try to reconnect.");
                        try {
                            client.set(initAndSubscribe(mqttClientConfig.getPlainTextPort(), mqttClientConfig.getBrokerHost(), vertx, options));
                        } catch (Exception e) {
                            log.error("failed to reconnect to emqx and re-subscribe, will retry in {}s", checkIntervalSeconds);
                        }
                    }
                });
    }

    private MqttClient initAndSubscribe(int port, String host, Vertx vertx, MqttClientOptions options) {
        if (client.get() != null && client.get().isConnected()) {
            return client.get();
        }

        io.vertx.mqtt.MqttClient newClient = io.vertx.mqtt.MqttClient.create(vertx, options);
        newClient.connect(port, host, ch -> {
            if (ch.succeeded()) {
                log.info("Connected to emqx: {}", mqttClientConfig.getBrokerHost() + ":" + mqttClientConfig.getPlainTextPort());

                newClient.publishHandler(this::handleOnMessageReceive).subscribe(Constants.MQTT.Topics.UPDATE_CON, mqttCleintSubscribeQos);

                newClient.publishHandler(this::handleOnMessageReceive).subscribe(Constants.MQTT.Topics.SET_R, mqttCleintSubscribeQos);

                newClient.publishHandler(this::handleOnMessageReceive).subscribe(Constants.MQTT.Topics.UPDATE_GET_DKEY, mqttCleintSubscribeQos);
            } else {
                log.error("Failed to connect to emqx: {}, reason: {}", mqttClientConfig.getBrokerHost() + ":" + mqttClientConfig.getPlainTextPort(), ch.cause().toString());
            }
        });

        newClient.subscribeCompletionHandler(h -> {
            log.info("Subscribe complete, levels" + h.grantedQoSLevels());
        });

        newClient.exceptionHandler(e -> {
            log.error("mqtt client encounterred errors: {}", e.getMessage());
        });

//        newClient.closeHandler(e ->{
//            log.error("close and reconnect to emqx");
//            try {
//                Thread.sleep(mqttClientConfig.getReconnectInterval());
//                initAndSubscribe(mqttClientConfig.getPlainTextPort(), mqttClientConfig.getBrokerHost(), vertx, options);
//            } catch (InterruptedException ex) {
//                throw new RuntimeException(ex);
//            }
//        });

        client.set(newClient);

        return newClient;
    }

    @Bean("mqttCustomClient")
    public MqttClient mqttCustomClient() {
        return initAndSubscribe(mqttClientConfig.getPlainTextPort(), mqttClientConfig.getBrokerHost(), vertx, options);
    }

    @Override
    public void afterPropertiesSet() {
        kafkaUploadPool = Schedulers.newBoundedElastic(mqttClientConfig.getUploadKafkaTopicExecPoolSize(), mqttClientConfig.getUploadKafkaTopicExecQueueSize(), "kafka-upload-pool");
        workerPool = Schedulers.newBoundedElastic(mqttClientConfig.getUploadKafkaTopicExecPoolSize(), mqttClientConfig.getUploadKafkaTopicExecQueueSize(), "worker-pool");
    }

    public Mono<ProductInfoInCache> fetchProductInfoInCaches(String domain, String subDomain) {
        return Mono.just(Optional.ofNullable(productInfoInCaches.getFromCache(String.format("%s-%s", domain, subDomain)))
                .orElse(ProductInfoInCache.builder().build())).zipWhen(inCache -> {
            if (0 != inCache.getMsgTransType()) {
                return Mono.just(inCache);
            } else {
                return Mono.just(productDao.findProductInfo(domain, subDomain)).or(Mono.just(new ProductInfo())).zipWhen(productInfo -> {
                    if (productInfo != null && productInfo.getMsgTransType() != 0) {
                        return Mono.just(ProductInfoInCache.builder()
                                .msgTransType(productInfo.getMsgTransType())
                                .oneMachineOneSecret(productInfo.getOneMachineOneSecret())
                                .communication(productInfo.getCommunication())
                                .deviceUplinkUrl(Optional.ofNullable(productInfo.getDeviceUplinkUrl()).orElse(""))
                                .build()).zipWhen(newInfo -> {
                            productInfoInCaches.saveToCache(String.format("%s-%s", domain, subDomain), newInfo);
                            return Mono.just(newInfo);
                        }, (x, y) -> y);
                    }
                    return Mono.just(ProductInfoInCache.builder().build());
                }, (x, y) -> y);
            }
        }, (x, y) -> y);
    }

    public Mono<AesKeyInfoInCache> fetchAesKeyInfoInCaches(String domain) {
        return Mono.just(Optional.ofNullable(aesKeyInfoInCaches.getFromCache(domain))
                .orElse(AesKeyInfoInCache.builder().build())).zipWhen(aesKeyInfoInCache -> {
            if (StringUtils.isEmpty(aesKeyInfoInCache.getAccess_key())) {
                DataSourceKey.use(Constants.DS.PLATFORM);
                return Mono.just(RowUtil.toEntity(Db.selectOneBySql(Constants.SQL_TEMP, domain), AesKeyInfoInCache.class)).zipWhen(aesInfo -> {
                    if (aesInfo != null) {
                        aesKeyInfoInCaches.saveToCache(domain, aesInfo);
                        return Mono.just(aesInfo);
                    }
                    return Mono.just(aesKeyInfoInCache);
                }, (x, y) -> y);
            } else {
                return Mono.just(aesKeyInfoInCache);
            }
        }, (x, y) -> y);
    }


    public boolean send(String topic, Buffer payload) {
        client.get().publish(
                topic,
                payload,
                MqttQoS.valueOf(mqttCleintPushQos),
                false,
                false,
                z -> log.info("publish msg: {} to topic: {}", payload, topic));

        return true;
    }

    public void handleOnMessageReceive(io.vertx.mqtt.messages.MqttPublishMessage msg) {
        String topicName = msg.topicName();
        log.info("Receivenew  MQTT msg-> Topic: {}, Payload bytes: {}", topicName, msg.payload().getBytes());
        String token = DkeyTool.generateRandomKey(8);
        String[] s = topicName.split("/");

        String majorDomainId = "";
        String subDomainId = "";
        String physicalId = "";

        MqttDeviceUploadMsg msgToKafka = MqttDeviceUploadMsg.builder()
                .build();

        if (topicName.contains("device_offline")) {
            majorDomainId = s[1];
            subDomainId = s[2];
            physicalId = s[3];

            msgToKafka.setDomain(majorDomainId);
            msgToKafka.setSubDomain(subDomainId);
            msgToKafka.setWifiId(physicalId);
            msgToKafka.setEvent(Constants.MQTT.API.Api_On_Deviceline);

            Map<String, Integer> data = new HashMap<>();
            data.put("status", Constants.STATUS_OFFLINE);
            msgToKafka.setPayload(JSONUtil.toJsonStr(data));
            msgToKafka.setOriginMsgType(Constants.MQTT.API.Api_On_Deviceline);
            msgToKafka.setPayloadFormat("");
        } else {
            majorDomainId = s[0];
            subDomainId = s[1];
            physicalId = s[2];

            msgToKafka.setDomain(majorDomainId);
            msgToKafka.setSubDomain(subDomainId);
            msgToKafka.setWifiId(physicalId);

            String encryptedPayload = java.util.Base64.getEncoder().encodeToString(msg.payload().getBytes());
            msgToKafka = MqttDeviceUploadMsg.builder()
                    .domain(majorDomainId)
                    .subDomain(subDomainId)
                    .wifiId(physicalId)
                    .payload(encryptedPayload)
                    .payloadFormat("Base64Payload")
                    .token(token)
                    .build();

            switch (s[4]) {
                case Constants.MQTT.API.Api_On_Update_Con -> {
                    msgToKafka.setOriginMsgType(Constants.MQTT.API.Api_On_Update_Con);
                    msgToKafka.setEvent(Constants.MQTT.API.Api_On_Notication);
                    msgToKafka.setRespPayload("{\"status\":200}");
                    String topic = String.format("%s/%s/%s/queueMode/updateConR/", majorDomainId, subDomainId, physicalId);

                    send(topic, Buffer.buffer("{\"status\":200}"));
                }
                case Constants.MQTT.API.Api_On_Update_SetR -> {
                    msgToKafka.setOriginMsgType(Constants.MQTT.API.Api_On_Update_SetR);
                    msgToKafka.setEvent(Constants.MQTT.API.Api_On_Notication);
                }
                case Constants.MQTT.API.Api_On_Update_Ver -> {
                    msgToKafka.setOriginMsgType(Constants.MQTT.API.Api_On_Deviceline);
                    msgToKafka.setEvent(Constants.MQTT.API.Api_On_Deviceline);
                    msgToKafka.setStatus(Constants.STATUS_ONLINE);

                    PubVerMsg pubVerMsg = PubVerMsg.builder().status(200).build();
                    String topic = String.format("%s/%s/%s/queueMode/updateVerR/", majorDomainId, subDomainId, physicalId);
                    String pubVerMsgJson = JSONUtil.toJsonStr(pubVerMsg);

                    send(topic,
                            Buffer.buffer(pubVerMsgJson));
                }
                case Constants.MQTT.API.Api_On_Update_Get_Dkey -> {
                    msgToKafka.setOriginMsgType(Constants.MQTT.API.Api_On_Update_Get_Dkey);
                    msgToKafka.setStatus(Constants.STATUS_ONLINE);

                    GenerateAesKeyAndFetchPublicKeyResp generateAesKeyAndFetchPublicKeyResp = mqttClients
                            .generateAesKeyAndFetchPublicKey(majorDomainId, subDomainId, physicalId).block();
                    String encryptedDkey = java.util.Base64.getEncoder().encodeToString(AESEcbEncryptUtil.aesEcbEncryptToBase64(generateAesKeyAndFetchPublicKeyResp.getPubkey()
                            , generateAesKeyAndFetchPublicKeyResp.getKey().getBytes(StandardCharsets.UTF_8)).getBytes(StandardCharsets.UTF_8));
                    String topic = String.format("%s/%s/%s/queueMode/updateGetDkeyR/", majorDomainId, subDomainId, physicalId);

                    send(topic,
                            Buffer.buffer(encryptedDkey));

                }
            }

            final MqttDeviceUploadMsg msgToKafkaFinal = msgToKafka;

            Mono.zip(
                            fetchProductInfoInCaches(majorDomainId, subDomainId),
                            fetchAesKeyInfoInCaches(majorDomainId)
                    ).zipWhen(tuple -> {
                        msgToKafkaFinal.setType(Constants.MQTT.MsgTransTypeMap.get(tuple.getT1().getMsgTransType()));
                        if (msgToKafkaFinal.getType().equals(Constants.MQTT.MsgTransTypes.All) || msgToKafkaFinal.getType().equals(Constants.MQTT.MsgTransTypes.TT)) {
                            msgToKafkaFinal.setDeviceUplinkUrl(tuple.getT1().getDeviceUplinkUrl());
                            msgToKafkaFinal.setAesKey("fbdd5c3940c6646280f6cbba5daa0f36");
                            msgToKafkaFinal.setSecretKey("c8b777ed40f289648059c2f070964882");

                            msgToKafkaFinal.setAesKey(tuple.getT2().getAccess_key());
                            msgToKafkaFinal.setSecretKey(tuple.getT2().getSecret_key());
                            msgToKafkaFinal.setDeveloper_id(tuple.getT2().getDeveloper_id());
                        }

                        return Mono.just(msgToKafkaFinal);
                    }, (x, y) -> y)
                    .flatMap(z -> {
                        reactiveKafkaSender.sendToKafkaReactiveShard(mqttClientConfig.getUploadKafkaTopic(), z.getWifiId(), JSONUtil.toJsonStr(z), kafkaUploadPool);
                        return Mono.just(true);
                    })
                    .subscribeOn(workerPool)
                    .subscribe();
        }
    }

    public Mono<Boolean> notifyWifiOnlineToHitachi(Mono<NotifyWifiOnlineToHitachiReq> req) {
        return req.flatMap(r -> fetchProductInfoInCaches(String.valueOf(r.getMajorDomainId()), String.valueOf(r.getSubDomainId()))
                .flatMap(res -> {
                            if (StringUtils.isNotEmpty(res.getDeviceUplinkUrl()) && res.getDeviceUplinkUrl().startsWith("http")) {
                                String baseUrl;
                                if (res.getDeviceUplinkUrl().contains("?")) {
                                    baseUrl = res.getDeviceUplinkUrl().split("\\?")[0] + Constants.MQTT.API.Api_On_Deviceline;
                                } else {
                                    baseUrl = res.getDeviceUplinkUrl() + Constants.MQTT.API.Api_On_Deviceline;
                                }
                                log.info("baseUrl: {}", baseUrl);
                                return webClient.post()
                                        .uri(baseUrl)
                                        .body(req, NotifyWifiOnlineToHitachiReq.class)
                                        .retrieve()
                                        .bodyToMono(String.class)
                                        .doOnSuccess(res1 -> {
                                            log.info("succeed to notifyWifiOnlineToHitachi {} for wifiId {}", r.getPayload(), r.getPhysicalId());
                                        })
                                        .doOnError(err -> log.error("succeed to notifyWifiOnlineToHitachi {} for wifiId {}", r.getPayload(), r.getPhysicalId(), err))
                                        .zipWhen(s -> Mono.just(s.contains("success")), (x, y) -> y);
                            } else {
                                log.error("no hitachi DeviceUplinkUrl configured for domain: {}, subdomain: {}, will not notify hitachi online msg anymore.",
                                        r.getMajorDomainId(), r.getSubDomainId());
                                return Mono.just(false);
                            }
                        }
                ).subscribeOn(workerPool));

    }
}
